package org.lpw.wormhole.queue;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.lpw.wormhole.element.Element;
import org.lpw.wormhole.util.Kit;

/**
 * @author lpw
 */
public class LocalQueueImpl
    implements LocalQueue, Comparator<QueueProcessor>
{
    protected final Kit kit;
    protected final Map<String, List<QueueProcessor>> processors;
    protected final ExecutorService threadPool;

    public LocalQueueImpl(Kit kit)
    {
        this.kit = kit;
        processors = Collections.synchronizedMap(new HashMap<String, List<QueueProcessor>>());
        threadPool = Executors.newFixedThreadPool(kit.getConfigurer().getQueueMaxThread());
    }

    @Override
    public void addProcessor(QueueProcessor processor)
    {
        if(processor==null)
            return;

        List<QueueProcessor> list = processors.get(processor.getQueueKey());
        if(list==null)
            list = Collections.synchronizedList(new ArrayList<QueueProcessor>());
        list.add(processor);
        Collections.sort(list, this);
        processors.put(processor.getQueueKey(), list);
    }

    @Override
    public void push(final Element element)
    {
        if(element==null||processors.isEmpty())
            return;

        final List<QueueProcessor> list = processors.get(element.getKey());
        if(list==null||list.isEmpty())
            return;

        threadPool.execute(new Runnable()
        {
            @Override
            public void run()
            {
                Map<Object, Object> context = new HashMap<Object, Object>();
                for(QueueProcessor processor : list)
                    processor.process(context, element.getValue());
            }
        });
    }

    @Override
    public void destory()
    {
        threadPool.shutdown();
    }

    @Override
    public int compare(QueueProcessor o1, QueueProcessor o2)
    {
        return o1.getProcessSort()-o2.getProcessSort();
    }
}
